package com.example.sangong.kafkacase;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

import java.util.List;
import java.util.Map;

public class MyPartitioner implements Partitioner {
    //为每个ProduceRecord计算分区号
    // 根据key的hashCode() % 分区数
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        //获取主题的分区数
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);

        int numPartitions = partitions.size();
        System.err.println("分区执行--");
        return (key.hashCode() & Integer.MAX_VALUE) % numPartitions;
    }

    // Producer执行close()方法时调用
    @Override
    public void close() {

    }

    // 从Producer的配置文件中读取参数,在partition之前调用
    @Override
    public void configure(Map<String, ?> configs) {


    }
}
